use serde_json::{Value, from_value};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel};
use tokio::sync::{Semaphore};
use tokio::time::{Duration, interval, sleep};

use binance_sdk::ws::ws_client::{WsClient, parse_market_event};
use binance_sdk::{
    config::Config,
    rest::{ExchangeInfo, OrderBook, client::BinanceRestClient},
    ws::market_types::MarketStreamEvent,
};
use common::constants::ORDER_BOOK_LEVEL;
use crate::app_config::AppConfig;
use crate::broadcast_handler::BroadcastHandler;

#[derive(Debug)]
pub enum QuoteMgrEvent {
    StreamEvent(MarketStreamEvent),
    RefreshExchangeInfo(ExchangeInfo),
    RefreshAllOrderBooks,
}

pub struct QuoteMgr {
    config: AppConfig,
    internal_tx: UnboundedSender<QuoteMgrEvent>,
    rest_client: Arc<BinanceRestClient>,
    exchange_info: Option<ExchangeInfo>,
    order_books: HashMap<String, OrderBook>,
    ws_client: Option<WsClient>,
    broadcast_handler: BroadcastHandler, // 新增
}

impl QuoteMgr {
    pub fn new(config: AppConfig) -> (Self, UnboundedReceiver<QuoteMgrEvent>) {
        let (tx, rx) = unbounded_channel();
        let rest_client = Arc::new(BinanceRestClient::new(config.sdk.clone()));
        let broadcast_handler = BroadcastHandler::new(config.clone());
        
        (
            Self {
                config,
                internal_tx: tx,
                rest_client,
                exchange_info: None,
                order_books: HashMap::new(),
                ws_client: None,
                broadcast_handler,
            },
            rx,
        )
    }

    pub async fn init(&mut self, mut internal_rx: UnboundedReceiver<QuoteMgrEvent>) {
        // 初始化WS连接
        let (ws_client, mut ws_rx) = WsClient::new(&format!("{}/stream", self.config.sdk.ws_base_url()),
                                                   self.config.sdk.proxy.clone()).await.unwrap();
        self.ws_client = Some(ws_client.clone());

        // 定义三个生产者
        let ws_tx = self.internal_tx.clone();
        let info_tx = self.internal_tx.clone();
        let refresh_tx = self.internal_tx.clone();

        // 启动WS消息处理器
        tokio::spawn(async move {
            while let Some(raw_msg) = ws_rx.recv().await {
                if let Ok(json) = from_value::<Value>(raw_msg) {
                    if let Some(event) = parse_market_event(json) {
                        let _ = ws_tx.send(QuoteMgrEvent::StreamEvent(event));
                    }
                }
            }
        });

        // 启动ExchangeInfo定时器
        tokio::spawn({
            let rest = self.rest_client.clone();
            async move {
                let mut timer = interval(Duration::from_secs(600));
                loop {
                    if let Ok(info) = rest.get_exchange_info().await {
                        let _ = info_tx.send(QuoteMgrEvent::RefreshExchangeInfo(info));
                    }
                    timer.tick().await;
                }
            }
        });

        // 启动OrderBook刷新定时器
        tokio::spawn(async move {
            let mut timer = interval(Duration::from_secs(15));
            loop {
                let _ = refresh_tx.send(QuoteMgrEvent::RefreshAllOrderBooks);
                timer.tick().await;
            }
        });

        // 启动消费者
        self.broadcast_handler.spawn_consumer();
        
        // 主事件循环
        let mut qm = self.clone_shallow();
        tokio::spawn(async move {
            while let Some(event) = internal_rx.recv().await {
                qm.handle_event(event).await;
            }
        });
    }

    fn clone_shallow(&self) -> Self {
        Self {
            config: self.config.clone(),
            broadcast_handler: self.broadcast_handler.clone(),
            internal_tx: self.internal_tx.clone(),
            rest_client: self.rest_client.clone(),
            exchange_info: self.exchange_info.clone(),
            order_books: self.order_books.clone(), // 注意这里只是拷贝，实际后续可用 Arc<Mutex> 优化
            ws_client: self.ws_client.clone(),
        }
    }

    fn init_orderbooks(&self, symbols: Vec<String>) {
        // 20并发 × 1秒间隔 = 1200权重/分钟 (刚好打满)
        let concurrency_limit = Arc::new(Semaphore::new(18));
        let internal_tx = self.internal_tx.clone();
        let rest_client = self.rest_client.clone();
    
        for symbol in symbols.iter() {
            let permit = concurrency_limit.clone();
            let symbol = symbol.clone();
            let internal_tx = internal_tx.clone();
            let rest_client = rest_client.clone();
    
            tokio::spawn(async move {
                let _permit = permit.acquire_owned().await.unwrap();
                let mut retry_count = 0;
                let max_retry = 2; // 减少重试次数
    
                loop {
                    match rest_client.get_order_book(&symbol, Some(ORDER_BOOK_LEVEL)).await {
                        Ok(order_book) => {
                            let event = MarketStreamEvent::OrderBook(order_book);
                            let _ = internal_tx.send(QuoteMgrEvent::StreamEvent(event));
                            break;
                        }
                        Err(e) => {
                            retry_count += 1;
                            if retry_count >= max_retry {
                                break;
                            }
                            sleep(Duration::from_secs(1)).await; // 错误时也等待1秒
                        }
                    }
                }
                // 固定1秒间隔
                sleep(Duration::from_secs(1)).await;
                drop(_permit);
            });
        }
    }

    async fn handle_stream_event(&mut self, event: MarketStreamEvent) {
        match event {
            MarketStreamEvent::BookTicker(tick) => {}
            MarketStreamEvent::OrderBook(orderbook) => {
                self.order_books.insert(orderbook.symbol.clone(), orderbook.clone());
                self.broadcast_handler.send(MarketStreamEvent::OrderBook(orderbook));
            }
            MarketStreamEvent::DepthUpdate(depth) => {
                if let Some(book) = self.order_books.get_mut(&depth.symbol) {
                    book.apply_depth_update(depth);
                    self.broadcast_handler.send(MarketStreamEvent::OrderBook(book.clone()));
                }
            }
            _ => {}
        }
    }

    async fn handle_event(&mut self, event: QuoteMgrEvent) {
        match event {
            QuoteMgrEvent::RefreshExchangeInfo(info) => {
                println!("[QuoteMgr] Got ExchangeInfo: {} symbols", info.symbols.len());
                self.exchange_info = Some(info.clone());
                
                if let Some(ws) = &self.ws_client {
                    let symbols = info.get_trading_symbols();
                    let topics = symbols.iter()
                        .map(|s| format!("{}@depth@100ms", s.to_lowercase()))
                        .collect::<Vec<_>>();
                    let _ = ws.subscribe_all(&topics).await;
                }
            }
            
            QuoteMgrEvent::RefreshAllOrderBooks => {
                println!("[QuoteMgr] RefreshAllOrderBooks");
                if let Some(info) = &self.exchange_info {
                    self.init_orderbooks(info.get_trading_symbols());
                }
            }
            
            QuoteMgrEvent::StreamEvent(evt) => {
                self.handle_stream_event(evt).await;
            }
        }
    }
}
